/**
 * 
 */
package com.concurrent.execute.threadPool;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.RunnableFuture;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.concurrent.execute.AsExecuteResult;


/**
 * 原始的java线程池是没有最大线程，最小线程区分的，在queue满之前，都是采用corePoolSize数量的线程运行，queue满之后会增加线程，
 * 直到maximunPoolSize数量的线程，因此提供哦功能动态线程池，动态的增加，减少线程数量。
 * 
 * @author liaokucheng
 * @date 2016年11月10日
 * 
 */
public class DynamicThreadPool {

	private static Logger log = LoggerFactory.getLogger(DynamicThreadPool.class);

	private volatile ThreadFactory threadFactory;

	/**
	 * Timeout in nanoseconds for idle threads waiting for work. Threads use this
	 * timeout when there are more than corePoolSize present or if
	 * allowCoreThreadTimeOut. Otherwise they wait forever for new work.
	 */
	private volatile long keepAliveTime;

	private TimeUnit unit;

	/**
	 * 最小线程数
	 */
	private volatile int minPoolSize;
	/**
	 * 最大线程数
	 */
	private volatile int maxPoolSize;
	/**
	 * 当前线程数
	 */
	private final AtomicInteger currentPoolSize = new AtomicInteger(0);
	/**
	 * 任务数量
	 */
	private final AtomicInteger taskNum = new AtomicInteger(0);

	/**
	 * 任务队列
	 */
	private final BlockingQueue<Runnable> workQueue;
	/**
	 * 0，当前线程直接执行，1 放弃
	 */
	private RejectedExecutionHandler reject;

	public DynamicThreadPool(int minPoolSize, int maxPoolSize, long keepAliveTime, TimeUnit unit,
			BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler reject) {
		if (minPoolSize < 0 || maxPoolSize <= 0 || maxPoolSize < minPoolSize || keepAliveTime < 0)
			throw new IllegalArgumentException();
		if (workQueue == null || threadFactory == null)
			throw new NullPointerException();
		this.minPoolSize = minPoolSize;
		this.maxPoolSize = maxPoolSize;
		this.workQueue = workQueue;
		this.keepAliveTime = keepAliveTime;
		this.unit = unit;
		this.threadFactory = threadFactory;
		this.reject = reject;
	}

	/**
	 * 状态，-1 关闭中，-2 已关闭,1未运行，2运行
	 */
	private AtomicInteger status = new AtomicInteger(1);

	private boolean allowIncreate() {
		while (status.get() > 0) {
			int value = currentPoolSize.get();
			if (value < this.maxPoolSize && taskNum.get() > value) {
				// 当前线程小于最大线程数，当前线程数加1,，增加成功返回true，否则循环执行
				if (currentPoolSize.compareAndSet(value, value + 1)) {
					return true;
				} else {
					continue;
				}
			} else {
				return false;
			}
		}
		return false;

	}

	/**
	 * 允许减少
	 * 
	 * @return
	 */
	private boolean allowDecrement() {
		while (status.get() > 0) {
			int value = currentPoolSize.get();
			if (value > this.minPoolSize && taskNum.get() < value) {
				if (currentPoolSize.compareAndSet(value, value - 1)) {
					return true;
				}
			} else {
				return false;
			}
		}
		return true;
	}

	final void reject(Runnable command) {
		reject.rejectedExecution(command, this);
	}

	private boolean addWork(Runnable command) {
		try {
			Worker worker = new Worker(command);
			Thread thread = threadFactory.newThread(worker);
			thread.start();
			return true;
		} catch (Error e) {
			log.error("", e);
			return false;
		}
	}

	public void execute(Runnable command) {
		if (null != command) {
			// 执行任务
			taskNum.incrementAndGet();
			if (allowIncreate()) {
				// 运行增加执行线程

				// 添加新的执行者失败，，减少执行者数量
				if (!addWork(command)) {
					currentPoolSize.decrementAndGet();
				}
			} else {
				if (this.workQueue.offer(command)) {

				} else {
					reject(command);
				}
			}

		}

	}

	/**
	 * 执行者
	 * 
	 * @author lkc
	 *
	 */
	private class Worker extends Thread {

		Runnable command;

		private Worker(Runnable command) {
			this.command = command;
		}

		public void run() {
			while (true) {
				if (status.get() < 0) {
					return;
				}
				if (null != command) {
					execute(command);
				} else if (allowDecrement()) {
					break;

				}
				getTask();
			}
		}

		/**
		 * 获取任务
		 */
		private void getTask() {
			try {
				command = workQueue.poll(keepAliveTime, unit);
			} catch (Exception e) {
				log.error("", e);
			}
		}

		/**
		 * 执行
		 * 
		 * @param command
		 */
		private void execute(Runnable command) {
			try {
				command.run();
			} catch (Error e) {
				log.error("", e);
			} finally {
				taskNum.decrementAndGet();
			}
		}

	}

	public void shutdown() {
		status.set(-2);

	}

	public boolean isShutdown() {
		return this.status.get() == -2;
	}

	public boolean isTerminated() {
		return this.status.get() == -1;
	}

	/**
	 * Returns a {@code RunnableFuture} for the given runnable and default value.
	 *
	 * @param runnable
	 *            the runnable task being wrapped
	 * @param value
	 *            the default value for the returned future
	 * @param <T>
	 *            the type of the given value
	 * @return a {@code RunnableFuture} which, when run, will run the underlying
	 *         runnable and which, as a {@code Future}, will yield the given value
	 *         as its result and provide for cancellation of the underlying task
	 * @since 1.6
	 */
	protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
		return new AsExecuteResult<T>(runnable, value);
	}

	/**
	 * Returns a {@code RunnableFuture} for the given callable task.
	 *
	 * @param callable
	 *            the callable task being wrapped
	 * @param <T>
	 *            the type of the callable's result
	 * @return a {@code RunnableFuture} which, when run, will call the underlying
	 *         callable and which, as a {@code Future}, will yield the callable's
	 *         result as its result and provide for cancellation of the underlying
	 *         task
	 * @since 1.6
	 */
	protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
		return new AsExecuteResult<T>(callable);
	}

	/**
	 * @throws RejectedExecutionException
	 *             {@inheritDoc}
	 * @throws NullPointerException
	 *             {@inheritDoc}
	 */
	public Future<?> submit(Runnable task) {
		if (task == null)
			throw new NullPointerException();
		RunnableFuture<Void> ftask = newTaskFor(task, null);
		execute(ftask);
		return ftask;
	}

	/**
	 * @throws RejectedExecutionException
	 *             {@inheritDoc}
	 * @throws NullPointerException
	 *             {@inheritDoc}
	 */
	public <T> Future<T> submit(Runnable task, T result) {
		if (task == null)
			throw new NullPointerException();
		RunnableFuture<T> ftask = newTaskFor(task, result);
		execute(ftask);
		return ftask;
	}

	/**
	 * @throws RejectedExecutionException
	 *             {@inheritDoc}
	 * @throws NullPointerException
	 *             {@inheritDoc}
	 */
	public <T> Future<T> submit(Callable<T> task) {
		if (task == null)
			throw new NullPointerException();
		RunnableFuture<T> ftask = newTaskFor(task);
		execute(ftask);
		return ftask;
	}
}
